package Service

import Config.Constant
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import redis.clients.jedis.Jedis

import java.util.Properties
object Project_Anime_Data {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val proper = new Properties()
    proper.setProperty("bootstrap.servers", Constant.KAFKA_SERVER_BIND)
    proper.setProperty("group.id",Constant.KAFKA_GROUP_ID)
    val data=env
      .addSource(new FlinkKafkaConsumer[String](Constant.KAFKA_TOPIC,new SimpleStringSchema(),proper))
      .addSink(new UserRichSinkFunction)
    env.execute()
  }
  class UserRichSinkFunction extends RichSinkFunction[String] {
    var jedis:Jedis=_
    override def open(parameters: Configuration): Unit = {
      jedis=new Jedis(Constant.REDIS_BIND,Constant.REDIS_POST)
      jedis.select(Constant.REDIS_DATABASE)
    }
    override def invoke(in: String): Unit = {
      jedis.publish(Constant.Project_Anime_Data,in)
      jedis.rpush(Constant.Project_Anime_Data,in)
      println(in)
    }

    override def close(): Unit = {
      jedis.close()
    }
  }
}
